package com.nokia.position;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.output.LzoProtobufB64LineOutputFormat;
import com.twitter.elephantbird.mapreduce.output.LzoProtobufBlockOutputFormat;

public class PositionBinaryDriver {
	
	public void printUsage() {
			System.out
					.println("Usage: hadoop jar path/to/this.jar "
							+ this.getClass()
							+ " <input dir> <output dir>");

	}

	public int run(String[] args, Configuration conf)
			throws Exception {
		
		if (args.length != 2) {
			printUsage();
			return 1;
		}
		
		Job job = new Job(conf);
		job.setJobName("Positioning binary convertion");

		job.setJarByClass(getClass());
		job.setMapperClass(PositionBinaryMapper.class);
		job.setNumReduceTasks(0);

		job.setInputFormatClass(TextInputFormat.class);
		if (conf.get("proto.test.format", "B64Line").equals("Block")) {
			LzoProtobufBlockOutputFormat.setClassConf(Position.class,
					job.getConfiguration());
			job.setOutputFormatClass(LzoProtobufBlockOutputFormat.class);
		} else { // assume B64Line
			LzoProtobufB64LineOutputFormat.setClassConf(Position.class,
					job.getConfiguration());
			job.setOutputFormatClass(LzoProtobufB64LineOutputFormat.class);
		}

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		
		Configuration conf = new Configuration();
		args = new GenericOptionsParser(conf, args).getRemainingArgs();
		PositionBinaryDriver driver = new PositionBinaryDriver();
		System.exit(driver.run(args, conf));
	}
}
